use std::io::Read;
use std::net::SocketAddr;
use std::net::{TcpListener, TcpStream};
use std::str;
use std::sync::{Arc, Mutex};
use std::thread;

use tokio::io::{self, AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener as Listener;
use tokio::net::TcpStream as Streams;
use tokio::sync::mpsc as chan;

struct ServerEvent{

}
impl ServerEvent{
    pub fn new()->Self{
        ServerEvent{}
    }
    pub fn on_connect_event(&self,s:&mut tokio::net::TcpStream,addr:SocketAddr){
        println!(">>>>>>>>>");
    }
    pub fn on_disconnect_event(&self){
        println!("<<<<<<<<<");
    }
    pub async fn on_data_event(&self,s:&mut tokio::net::TcpStream,data:Vec<u8>){
        println!("==={:?}",data.bytes());
        client_send_package(s,data).await;
    }
}

// #[derive(Debug)]
// pub enum ServerEvent<'a> {
//     ConnectionEstablished(&'a mut tokio::net::TcpStream,SocketAddr),
//     ConnectionClosed(SocketAddr),
//     DataReceived(SocketAddr, Vec<u8>),
//     ErrorOccurred(SocketAddr, io::Error),
// }
pub struct TcpServer {}
impl TcpServer {
    pub fn new() -> TcpServer {
        TcpServer {}
    }
    pub fn run_server(addr: &str) {
        let conn = TcpListener::bind(addr);
        match conn {
            Ok(clients) => {
                thread::spawn(move || {
                    println!("服务器启动，监听 3000 端口");
                    for stream in clients.incoming() {
                        thread::spawn(move || {
                            handle_client(stream.unwrap());
                        });
                    }
                });
            }
            Err(err) => {
                println!("bind port fail,error:{}", err.to_string());
            }
        }
    }
}

fn handle_client(mut stream: TcpStream) {
    println!("Client Connect,Address={:?}", stream.peer_addr());
    let mut buf = [0; 128];
    loop {
        // 读取内容
        let result = stream.read(&mut buf);
        match result {
            Ok(len) => {
                if len == 0 {
                    println!("Client DisConnect,Address={:?}", stream.peer_addr());
                    break;
                }
                // 输出读取到的内容
                println!("read {} bytes: {:?}", len, str::from_utf8(&buf[..len]));
            }
            Err(err) => {
                println!("Client Error={:?}", err.to_string());
                break;
            }
        }
    }
}

fn vec_to_u16_be(v: &Vec<u8>) -> u16 {
    // 先将切片转换为 [u8; 4] 数组
    let mut arr = [0u8; 2];
    arr.copy_from_slice(&v[0..2]);
    u16::from_be_bytes(arr)
}
fn u16_to_vec_be(num: u16) -> Vec<u8> {
    let high = (num >> 8) as u8; // 取高8位
    let low = num as u8; // 取低8位
    vec![high, low]
}

async fn client_read_package(client: &mut Streams) -> std::io::Result<Vec<u8>> {
    let res = client.read_u16().await;
    return match res {
        Ok(len) => {
            let mut data = vec![0u8; len as usize];
            let cc = client.read_exact(&mut data).await;
            match cc {
                Ok(ll) => Ok(data),
                Err(err) => Err(err),
            }
        }
        Err(err) => Err(err),
    };
}

pub async fn client_send_package(client: &mut Streams, buf: Vec<u8>) -> bool {
    let len = buf.len();
    let mut temp = vec![0; len + 2];
    let ll = u16_to_vec_be(len as u16);
    temp.extend_from_slice(&ll);
    temp.extend_from_slice(&buf);
    let mut rst = false;
    let l = client.write_all(&temp).await;
    match l {
        Ok(_) => {
            rst = true;
        }
        Err(_) => {
            rst = false;
        }
    }
    return rst;
}

pub struct AsyncTcpServer {}
impl AsyncTcpServer {
    pub fn new() -> AsyncTcpServer {
        AsyncTcpServer {}
    }

    pub async fn run_server(self, addr: &str) {
        // 绑定端口
        let listener = Listener::bind(addr).await;
        match listener {
            Ok(clients) => {
                println!("服务器启动，监听 60000 端口");
                tokio::spawn(async move {
                    loop {
                        // 接受客户端连接
                        let result = clients.accept().await;
                        match result {
                            Ok(mut client) => {
                                // 为每个连接创建独立任务处理
                                tokio::spawn(async move {
                                    let events = ServerEvent::new();
                                    events.on_connect_event(&mut client.0,client.1);
                                    loop {
                                        let data = client_read_package(&mut client.0).await;
                                        match data {
                                            Ok(buf) => {
                                                events.on_data_event(&mut client.0,buf).await;
                                            }
                                            Err(err) => {
                                                break;
                                            }
                                        }
                                    }
                                    events.on_disconnect_event();
                                });
                            }
                            Err(err) => {
                                eprintln!("读取错误：{}", err.to_string());
                            }
                        }
                    }
                });
            }
            Err(err) => {
                println!("服务器启动失败，监听 60000 端口，错误={}", err.to_string());
            }
        }
    }
}
/////////////////////////////////////////////////////////////////////////////////
pub struct AsyncTcpClient {
    send_tx: chan::Sender<u8>,
    recv_tx: chan::Receiver<u8>,
    re_connect_max: u8,
}
impl AsyncTcpClient {
    pub fn new() -> AsyncTcpClient {
        let (tx, rx) = chan::channel::<u8>(100);
        AsyncTcpClient {
            send_tx: tx,
            recv_tx: rx,
            re_connect_max: 0, //等于0一直自动重连
        }
    }
    pub fn set_max_connect_num(&mut self, num: u8) {
        self.re_connect_max = num;
    }
    pub async fn connect(&mut self, addr: &str) {
        let mut count = 0;
        loop {
            // 连接服务器
            let stream = Streams::connect(addr).await;
            match stream {
                Ok(mut client) => {
                    println!(
                        "Connect Server OK,Addr={}",
                        client.peer_addr().unwrap().to_string()
                    );
                    let tx = self.send_tx.clone();
                    tokio::spawn(async move {
                        loop {
                            let data = client_read_package(&mut client).await;
                            match data {
                                Ok(buf) => {
                                    println!("Receive Data:{:?}", buf);
                                }
                                Err(err) => {
                                    print!("Client Read Error:{}", err.to_string());
                                    break;
                                }
                            }
                        }
                        tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
                        tx.send(0x01).await.unwrap(); // .expect("Send message Fail.");
                        count += 1;
                    });
                }
                Err(err) => {
                    println!("Connect Fail,Err:{}", err.to_string());
                    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
                    self.send_tx.send(0x01).await.expect("Send message Fail.");
                    count += 1;
                }
            }

            _ = self.recv_tx.recv().await;
            if count >= self.re_connect_max && self.re_connect_max > 0 {
                break;
            }
        }
    }
}
